package com.atguigu.flink0922.chapter12;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/12 13:59
 */
public class Flink01_Project_Product_TopN {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        // 1. 建立读取文件表
        tenv.executeSql(
            "create table user_behavior(" +
                "   user_id bigint, " +
                "   item_id bigint, " +
                "   category_id int, " +
                "   behavior string, " +
                "   ts bigint, " +
                "   event_time as to_timestamp(from_unixtime(ts, 'yyyy-MM-dd HH:mm:ss')), " +
                "   watermark for event_time as  event_time - interval '5' second " +
                ")with(" +
                "   'connector'='filesystem', " +
                "   'path'='input/UserBehavior.csv', " +
                "   'format'='csv')"
        );
        
        // 2.先计算每个商品的点击量  使用滑动窗口
        final Table t1 = tenv.sqlQuery(
            "select " +
                "   item_id, " +
                "   hop_end(event_time, interval '10' minute, interval '1' hour) w_end, " +
                "   count(*) item_count " +
                "from user_behavior " +
                "where behavior='pv' " +
                "group by hop(event_time, interval '10' minute, interval '1' hour), item_id"
        );
        tenv.createTemporaryView("t1", t1);
        
        // 3. 每个点击量进行排名 over窗口
        final Table t2 = tenv.sqlQuery(
            "select " +
                "   *, " +
                "   row_number() over(partition by w_end order by item_count desc) rn " +
                "from t1"
        );
        tenv.createTemporaryView("t2", t2);
  
        // 3. 取top5
        final Table t3 = tenv.sqlQuery(
            "select " +
                " w_end, " +
                " item_id, " +
                " item_count, " +
                " rn rk " +
                "from t2 " +
                "where rn<=5"
        );
        // 4. 数据sink到mysql
        // 4.1. 建表, 关联mysql
        tenv.executeSql("create table hot_item(" +
                            "   w_end timestamp(3), " +
                            "   item_id bigint, " +
                            "   item_count bigint, " +
                            "   rk bigint, " +
                            "   PRIMARY KEY (w_end, rk) NOT ENFORCED)" +
                            "with(" +
                            "   'connector' = 'jdbc', " +
                            "   'url' = 'jdbc:mysql://hadoop162:3306/flink_sql?useSSL=false', " +
                            "   'table-name' = 'hot_item', " +
                            "   'username' = 'root', " +
                            "   'password' = 'aaaaaa' " +
                            ")");
        
        // 4.2把数据sink到输出表
        t3.executeInsert("hot_item");
    }
}
